package com.zxy.zxyrpc.registry;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.util.StrUtil;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.zxy.zxyrpc.config.RegistryConfig;
import com.zxy.zxyrpc.exception.RpcException;
import com.zxy.zxyrpc.model.ErrorCode;
import com.zxy.zxyrpc.model.ServiceMetaInfo;
import io.lettuce.core.*;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * Redis 注册中心
 *
 * @author zxy
 * @since 2025/5/18
 **/
public class RedisRegistry implements Registry {
    /**
     * 根节点（业务前缀）
     */
    private static final String ETCD_ROOT_PATH = "/rpc/";
    /**
     * 本机注册节点 key 集合（用于维护心跳续期）
     */
    private final Set<String> localRegisterNodeKeySet = new HashSet<>();
    /**
     * 注册服务缓存
     */
    private final RegistryServiceCache registryServiceCache = new RegistryServiceCache();
    /**
     * 监听的 key 集合（ConcurrentHashSet 防止并发冲突）
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();
    private RedisClient redisClient;
    private StatefulRedisConnection<String, String> connect;
    private RedisCommands<String, String> redisCommands;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        RedisURI redisURI = RedisURI.builder()
                .withHost("localhost")
                .withPort(6379)
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                .build();
        RedisClient redisClient1 = RedisClient.create(redisURI);
        StatefulRedisConnection<String, String> connect1 = redisClient1.connect();
        RedisCommands<String, String> redisCommands = connect1.sync();
        SetArgs setArgs = SetArgs.Builder.nx().ex(5);
        redisCommands.set("myName", "zxyyyds", setArgs);
        System.out.println(redisCommands.get("myName"));

        connect1.close();
        redisClient1.close();
    }

    @Override
    public void init(RegistryConfig registryConfig) {
        RedisURI redisURI = RedisURI.builder()
                .withHost("localhost")
                .withPort(6379)
                .withTimeout(Duration.of(10, ChronoUnit.SECONDS))
                .build();
        redisClient = RedisClient.create(redisURI);
        connect = redisClient.connect();
        redisCommands = connect.sync();
        heartBeat();
    }

    @Override
    public void registry(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // 设置要存储的键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        String jsonStr = JSONUtil.toJsonStr(serviceMetaInfo);
        // 键值对与租约关联
        SetArgs setArgs = SetArgs.Builder.ex(30);
        redisCommands.set(registerKey, jsonStr, setArgs);
        // 添加节点信息到本地缓存
        localRegisterNodeKeySet.add(registerKey);
    }

    @Override
    public void unRegistry(ServiceMetaInfo serviceMetaInfo) {
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        redisCommands.del(registerKey);
        localRegisterNodeKeySet.remove(registerKey);
    }

    // ？
    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 先走服务缓存
        List<ServiceMetaInfo> serviceMetaInfos = registryServiceCache.readCache(serviceKey);
        if (serviceMetaInfos != null)
            return serviceMetaInfos;
        // 前缀索引，后面一定要加 "/"
        String searchPrefix = ETCD_ROOT_PATH + serviceKey;
        try {
            // 前缀查询
            ScanArgs scanArgs = ScanArgs.Builder.matches(searchPrefix);
            // 初始化游标和结果列表
            List<String> keys = new ArrayList<>();
            // 执行 SCAN 命令进行迭代查询
            KeyScanCursor<String> scan = redisCommands.scan(scanArgs);
            keys.addAll(scan.getKeys());

            // 解析服务列表信息
            // watch(key);
            List<ServiceMetaInfo> serviceMetaInfoList = keys.stream()
                    .map(v -> JSONUtil.toBean(v, ServiceMetaInfo.class))
                    .collect(Collectors.toList());
            // 写入服务缓存
            registryServiceCache.writeCache(serviceKey, serviceMetaInfoList);
            return serviceMetaInfoList;
        } catch (Exception e) {
            throw new RpcException(ErrorCode.OPERATION_ERROR, "获取服务列表失败");
        }
    }

    @Override
    public void destroy() {
        System.out.println("当前服务节点下线");
        // 遍历本服务下，所有的节点实例
        for (String key : localRegisterNodeKeySet) {
            try {
                redisCommands.del(key);
            } catch (Exception e) {
                throw new RpcException(ErrorCode.OPERATION_ERROR, key + "节点下线失败");
            }
        }
        // 释放锁（是否需要锁检查啊）
        if (connect != null)
            connect.close();
        if (redisClient != null)
            redisClient.close();
    }

    @Override
    public void heartBeat() {
        // 10秒续期
        CronUtil.schedule("*/10 * * * * *", (Task) () -> {
            // todo 遍历所有注册的服务节点。问题：时间统一可能并不好(前缀索引)
            for (String key : localRegisterNodeKeySet) {
                try {
                    String value = redisCommands.get(key);
                    // 该节点已过期（需要服务启动，并重新注册）
                    if (StrUtil.isBlank(value))
                        continue;
                    // 节点未过期，重新注册（续期）
                    ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                    registry(serviceMetaInfo);
                } catch (Exception e) {
                    throw new RpcException(ErrorCode.OPERATION_ERROR, key + " 续签失败");
                }
            }
        });
        // 支持秒级别的定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }

    // todo 待实现
    @Override
    public void watch(String serviceKey, String serviceNodeKey) {
    }
}
